package com.example.spider.cluster.redis;

import lombok.Data;
import lombok.NoArgsConstructor;
import org.redisson.api.RBitSet;
import org.redisson.api.RMap;
import org.redisson.api.RedissonClient;
import org.shoulder.batch.progress.BatchProgressRecord;
import org.shoulder.batch.progress.Progress;
import org.shoulder.batch.progress.ProgressAble;
import org.shoulder.batch.progress.ProgressStatusEnum;
import org.shoulder.core.exception.CommonErrorCodeEnum;
import org.shoulder.core.util.AssertUtils;

import java.io.Serializable;
import java.time.Duration;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.Map;
import java.util.Optional;
import java.util.function.BiConsumer;


/**
 * 批量处理进度并发模型
 * <p>
 * 额外附带了子任务幂等能力，适用于已经确定子任务数目的程序，且子任务id是从0连续有序的，只能记录完成数，不能记录失败数目
 * 线程不安全：建议通过加锁保证
 * <p>
 * 修改状态未带锁: 方法非原子方法
 *
 * @author lym
 */
@Data
@NoArgsConstructor
@javax.annotation.concurrent.NotThreadSafe
public class RedisFixedNumProgress implements Serializable, Progress {

    private static final long serialVersionUID = 1L;

    static final String ext_suffix = ":ext";
    static final String bitset_suffix = ":bs";
    static String CACHE_KEY_PREFIX = "";
    static String ATRR_KEY_PREFIX = "";

    private RedissonClient redissonClient;

    /**
     * 任务标识
     */
    private String id;

    /**
     * 额外新增
     * callbackStatus: (none)、Done
     */
    private RMap<String, String> attributeMap;

    private RBitSet rBitSet;

    private RMap<String, String> extMap;

    // note 【需要处理幂等】
    private BiConsumer<String, Progress> onFinishCallback = (x, y) -> {
    };

    public static void setCacheKeyPrefix(String cacheKeyPrefix) {
        RedisFixedNumProgress.CACHE_KEY_PREFIX = cacheKeyPrefix;
    }

    public static void setAtrrKeyPrefix(String atrrKeyPrefix) {
        RedisFixedNumProgress.ATRR_KEY_PREFIX = atrrKeyPrefix;
    }

    public static RedisFixedNumProgress of(String taskId, RedissonClient redissonClient) {
        AssertUtils.notBlank(taskId, CommonErrorCodeEnum.ILLEGAL_STATUS);
        taskId = taskId.replaceFirst(CACHE_KEY_PREFIX, "");
        RedisFixedNumProgress progress = new RedisFixedNumProgress();
        progress.id = taskId;
        progress.redissonClient = redissonClient;
        progress.attributeMap = redissonClient.getMap(CACHE_KEY_PREFIX + taskId);
        progress.rBitSet = redissonClient.getBitSet(ATRR_KEY_PREFIX + taskId + bitset_suffix);
        progress.extMap = redissonClient.getMap(ATRR_KEY_PREFIX + taskId + ext_suffix);
        return progress;
    }


    public void start() {
        AssertUtils.notBlank(id, CommonErrorCodeEnum.ILLEGAL_STATUS);
        attributeMap.putIfAbsent("startTime", System.currentTimeMillis() + "");
        attributeMap.putIfAbsent("status", "1");
        attributeMap.putIfAbsent("alreadyFinishedAtStart", getFinishedNum() + "");
    }

    public void failStop() {
        attributeMap.put("status", "2");
        attributeMap.put("stopTime", System.currentTimeMillis() + "");
    }

    public void finish() {
        attributeMap.put("status", "3");
        attributeMap.put("stopTime", System.currentTimeMillis() + "");

    }

    public boolean hasFinish() {
        checkFinished();
        return getStatus() > ProgressStatusEnum.RUNNING.getCode();
    }

    public int getStatus() {
        return Optional.ofNullable(attributeMap.get("status"))
                .map(Integer::parseInt)
                .orElse(0);
    }

    public int getTotal() {
        return Optional.ofNullable(attributeMap.get("total"))
                .map(Integer::parseInt)
                .orElse(0);
    }

    public void setTotal(int total) {
        attributeMap.put("total", total + "");
    }

    public int getAlreadyFinishedAtStart() {
        return Optional.ofNullable(attributeMap.get("alreadyFinishedAtStart"))
                .map(Integer::parseInt)
                .orElse(0);
    }

    public long getStopTime() {
        return Optional.ofNullable(attributeMap.get("stopTime"))
                .map(Long::parseLong)
                .orElse(0L);
    }

    public long getStartTime() {
        return Optional.ofNullable(attributeMap.get("startTime"))
                .map(Long::parseLong)
                .orElse(0L);
    }

    public long getFinishedNum() {
        return rBitSet.cardinality();
    }

    /**
     * 已经花费的时间
     *
     * @return 已经花费的时间
     */
    public long calculateProcessedTime() {
        if (getStatus() == ProgressStatusEnum.WAITING.getCode()) {
            return 0;
        }
        checkFinished();
        long end = Optional.of(getStopTime()).orElse(System.currentTimeMillis());
        return end - getStartTime();
    }

    /**
     * 估算当前进度
     *
     * @return 进度
     */
    public float calculateProgress() {
        if (hasFinish()) {
            return 1;
        }
        // 即将完成 99%
        long process = getFinishedNum();
        long totalNum = getTotal();
        return process == totalNum ? 0.999F : process / (float) totalNum;
    }

    /**
     * 估算剩余所需时间
     *
     * @return 剩余所需时间 ms
     */
    public long calculateTimeLeft() {
        if (hasFinish()) {
            return 0L;
        }
        long processedNum = getFinishedNum();
        int totalNum = getTotal();
        int alreadyFinishedAtStart = getAlreadyFinishedAtStart();
        if (processedNum - alreadyFinishedAtStart <= 0) {
            // 默认 99 天
            return Duration.ofDays(99).toMillis();
        }
        return (calculateProcessedTime() / (processedNum - alreadyFinishedAtStart) * (totalNum - processedNum));
    }

    @Override
    public String toString() {
        return "BatchProgress{" +
                "id='" + id + '\'' +
                ", total=" + getTotal() +
                ", processed=" + getFinishedNum() +
                ", startTime=" + LocalDateTime.ofInstant(Instant.ofEpochMilli(getStartTime()), ZoneId.systemDefault()) +
                ", stopTime=" + LocalDateTime.ofInstant(Instant.ofEpochMilli(getStopTime()), ZoneId.systemDefault()) +
                ", status=" + getStatus() +
                '}';
    }

    public BatchProgressRecord toRecord() {
        // set 顺序按照不易变化 -> 易变化顺序设置，为了逻辑严谨部分字段内容、顺序单独处理
        BatchProgressRecord record = new BatchProgressRecord();
        record.setId(id);
        long startTime = getStartTime();
        record.setStartTime(startTime == 0 ? null : LocalDateTime.ofInstant(Instant.ofEpochMilli(startTime), ZoneId.systemDefault()));
        record.setAlreadyFinishedAtStart(0);
        record.setStatus(getStatus());
        long stopTime = getStopTime();
        record.setStopTime(stopTime == 0 ? null : LocalDateTime.ofInstant(Instant.ofEpochMilli(stopTime), ZoneId.systemDefault()));
        record.setFailNum(0);
        record.setSuccessNum((int) getFinishedNum());
        record.setProcessed(record.getSuccessNum() + record.getFailNum());
        record.setTotal(getTotal());
        record.setExt((Map) extMap.readAllMap());
        return record;
    }

    @Override
    public BatchProgressRecord getBatchProgress() {
        return toRecord();
    }

    @Override
    public void finishPart(int partIndex) {
        this.rBitSet.set(partIndex);
        //
        if (hasFinish()) {
            // 调用onFinished
            onFinished(id, this);
        }
    }

    @Override
    public void onFinished(String id, ProgressAble task) {
        onFinishCallback.accept(id, (Progress) task);
    }

    private void checkFinished() {
        if (getFinishedNum() == getTotal()) {
            finish();
        }
    }

}
